package com.mango.ch03;

import com.mango.HDFSTools.HDFSAPI;
import org.apache.hadoop.fs.Path;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.broadcast.Broadcast;
import scala.Tuple2;

import java.io.IOException;
import java.net.URISyntaxException;
import java.util.*;
import java.util.Map.Entry;

/**
 * 使用spark高层api实现topN ！！！结果数据为写入hdfs中
 * 
 * @author Mango
 *
 */
public class SparkTopNJob {
    static Path inputPath = new Path("d:/HadoopData/SparkTopNJob/input");
	static Path outputPath = new Path("/SparkTopNJob/output");
	static int TOPN = 10;

	@SuppressWarnings("serial")
	public static void main(String[] args) {
		// TODO Auto-generated method stub
		// 确保有正确的输入输出
		HDFSAPI hdfs;
		try {
			hdfs = new HDFSAPI();
			hdfs.createDirectory(inputPath);
			hdfs.delDirectory(outputPath);
		} catch (IOException | InterruptedException | URISyntaxException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			System.out.println("输入输出文件路径有误");
		}
		// 创建于spark master的链接
		SparkConf sconf = new SparkConf();
		sconf.setMaster("local");
		JavaSparkContext jsc = new JavaSparkContext(sconf);
		// 动态过去需要指定的topN值
		Broadcast<Integer> topN = jsc.broadcast(TOPN);
		// 从hdfs读取输入文件并且创建第一个RDD
		System.out.println("input path  : " + inputPath.toString());
		JavaRDD<String> lines = jsc.textFile(inputPath.toString(), 1);
		// 创建一组tuple2
		JavaPairRDD<String, Integer> pairs = lines.mapToPair(new PairFunction<String, String, Integer>() {

			@Override
			public Tuple2<String, Integer> call(String t) throws Exception {
				// TODO Auto-generated method stub
				String[] tokens = t.split(",");// cat24,123
				Tuple2<String, Integer> tuple2 = new Tuple2<String, Integer>(tokens[0], Integer.valueOf(tokens[1]));
				return tuple2;
			}
		});
		pairs.saveAsTextFile(outputPath.toString() + "/1");
		// 为各个输入分区创建一个本地top10(即将各个partition中的数据进行一次topN统计，创建本地topN）
		JavaRDD<SortedMap<Integer, String>> partitions = pairs
				.mapPartitions(new FlatMapFunction<Iterator<Tuple2<String, Integer>>, SortedMap<Integer, String>>() {

					/**
					 * 
					 */
					private static final long serialVersionUID = 1L;

					@Override
					public Iterator<SortedMap<Integer, String>> call(Iterator<Tuple2<String, Integer>> t)
							throws Exception {
						// TODO Auto-generated method stub
						SortedMap<Integer, String> top10 = new TreeMap<>();// 定义一个treeMap来放置本地topN数据
						while (t.hasNext()) {
							Tuple2<String, Integer> item = t.next();
							top10.put(item._2, item._1);
							if (top10.size() > 10) {
								top10.remove(top10.firstKey());
							}
						}
						// return (Iterator<SortedMap<Integer, String>>)
						// Collections.singletonList(top10);
						return Collections.singletonList(top10).iterator();
					}
				});
		partitions.saveAsTextFile(outputPath.toString() + "/2");
		// 收集所有本地的top10并且创建最终的top10列表 (第一种处理方案)
		SortedMap<Integer, String> finaltop10 = new TreeMap<>();
		List<SortedMap<Integer, String>> alltop10 = partitions.collect();// 收集各个分区器的所有top10列表
		for (SortedMap<Integer, String> localtop10 : alltop10) {
			for (Entry<Integer, String> Entry : localtop10.entrySet()) {
				finaltop10.put(Entry.getKey(), Entry.getValue());
			}
			if (finaltop10.size() > topN.getValue())
				finaltop10.remove(finaltop10.firstKey());
		}
		// 输出最终的top10列表
		System.out.println("==========final top 10");
		for (Entry<Integer, String> entry : finaltop10.entrySet()) {
			System.out.println(entry.getKey() + "->>>" + entry.getValue());
		}
		// 第二种处理方案
		SortedMap<Integer, String> final_top10 = new TreeMap<>();
		final_top10 = partitions.reduce(
				// 前两个参数都是输入(各个分区中的LocaltopN列表)
				// 第三个参数是输出
				new Function2<SortedMap<Integer, String>, SortedMap<Integer, String>, SortedMap<Integer, String>>() {
					SortedMap<Integer, String> top10 = new TreeMap<>();

					@Override
					public SortedMap<Integer, String> call(SortedMap<Integer, String> v1, SortedMap<Integer, String> v2)
							throws Exception {
						// TODO Auto-generated method stub
						for (Entry<Integer, String> entry : v1.entrySet()) {
							top10.put(entry.getKey(), entry.getValue());
							if (top10.size() > topN.getValue())
								top10.remove(top10.firstKey());
						}
						for (Entry<Integer, String> entry : v2.entrySet()) {
							top10.put(entry.getKey(), entry.getValue());
							if (top10.size() > topN.getValue())
								top10.remove(top10.firstKey());
						}
						return top10;
					}
				});
		// 第二种方案输出最终的top10列表
		System.out.println("==========final top 10");
		for (Entry<Integer, String> entry : final_top10.entrySet()) {
			System.out.println(entry.getKey() + "->>>" + entry.getValue());
		}
		// 最终结果输出到文档
		jsc.close();
		System.exit(0);
	}

}
